#pragma once
#include <string>
#include "acl_cpp/lib_acl.hpp"
#include <memory>
#include "socket_session_acceptor.h"
#include "blacklist_filter.h"
#include "config.h"
#include "socket_session_auth.h"
#include "client_session.h"
#include "client_session_manager.h"
#include "subscribe_manager.h"

namespace acl
{
namespace mqtt
{
class acl_cpp_init_helper
{
public:
	acl_cpp_init_helper() {
		acl::acl_cpp_init();
	}
};

class mqtt_server
{
public:
	mqtt_server ()
		:aio_handle_(ENGINE_KERNEL)
	{
		static acl_cpp_init_helper inst;
	}
	bool bind_addr (const std::string &addr)
	{
		auto acceptor = 
			std::make_shared<socket_session_acceptor> (aio_handle_);
		if (acceptor->bind_addr (addr) == false)
			return false;
		acceptor->bind_accept_callback ([this] (socket_session && sess)
		{
			int max_connections = config::get_instance()
				.get_max_connection();
			if (max_connections > 0 &&
				connection_count_ > max_connections ) {
				sess.destory();
				return false;
			}
			handle_socket_session(std::move(sess));
			return true;
		});
		acceptors_.push_back (acceptor);
		return true;
	}
	void run ()
	{
		auto &timer_handle = client_session_manager::get_instance ()
			.get_timer_handle ();
		timer_handle.keep_timer (true);
		aio_handle_.set_timer (&timer_handle, 1000000);
		while (is_stop_ == false) {
			if(aio_handle_.check () == false)
				break;
		}
		aio_handle_.check ();
	}
	void stop ()
	{
		is_stop_ = true;
	}
	void enable_blacklist_filter()
	{
		blacklist_filter_ = new default_blacklist_filter();
	}
private:
	void handle_socket_session(socket_session && _socket_session)
	{
		if (blacklist_filter_)
		{
			std::string s = _socket_session.get_ip();
			if (!blacklist_filter_->check(s))
			{
				_socket_session.destory();
				return;
			}
		}
				
		auto ss_auth = std::make_shared<socket_session_auth>();
		ss_auth->auth_timeout(10);
		
		socket_sess_auths_.insert(socket_sess_auths_.end(), ss_auth);

		ss_auth->bind_socket_session(std::move(_socket_session))
		.bind_auth_failed_callback([this](socket_session &&ss){
			connection_count_--;
			ss.destory();
		})
		.bind_auth_succeed_callback([this](socket_session &&ss, 
											connect_msg &&msg, 
											packet_parser &&_parser) {

			socket_session_auth_secceed(std::move(ss), 
										std::move(msg), 
										std::move(_parser));
		}).run ();

		connection_count_++;
	}
	void socket_session_auth_secceed(socket_session &&sock_sess_,
									 connect_msg &&msg,
									 packet_parser &&_parser)
	{
		std::shared_ptr<client_session> client_sess;
		if(msg.connect_flags_.clean_session_) {
			client_sess = std::make_shared<client_session>();
			if(msg.clientid_.size()) {
				client_session_manager::get_instance()
					.delete_session(msg.clientid_);
				client_sess->bind_session_id(msg.clientid_);
			}
			else {
				std::string clientId = make_client_id();
				client_sess->bind_session_id(msg.clientid_);
			}
			client_sess->session_reuse (false);
		}
		else {
			client_sess = client_session_manager::get_instance()
				.get_session(msg.clientid_);
			if (client_sess == nullptr) {
				client_sess = std::make_shared<client_session> ();
				client_sess->session_reuse (false);
			}else
				client_sess->session_reuse (true);
			client_sess->bind_session_id(msg.clientid_);
		}
		client_session_manager::get_instance().add_session(client_sess);
		client_sess->
			bind_packet_parser (std::move (_parser)).
			bind_socket_session (std::move (sock_sess_)).
			ack_connect_ok (std::move (msg));
	}

	std::string make_client_id() 
	{
		std::stringstream buf;
		auto epoch = std::chrono::high_resolution_clock::now()
			.time_since_epoch().count();
		buf << config::get_instance().get_client_prefix() << epoch;
		return buf.str();
	}

private:
	int connection_count_ = 0;
	blacklist_filter * blacklist_filter_ = nullptr;
	bool is_stop_ = false;
	aio_handle aio_handle_;
	std::vector<std::shared_ptr<socket_session_acceptor>> acceptors_;
	std::list<std::shared_ptr<socket_session_auth>> socket_sess_auths_;
};
}
}
